package com.dx.chat.kafka.partition;

import com.dx.chat.kafka.core.KafkaConfig;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.kafka.core.KafkaAdmin;

import javax.annotation.PostConstruct;

/**
 * kafka初始化topic及partition
 *
 * @author wcx
 * @date 2023/6/25
 */
@RequiredArgsConstructor
@Configuration
public class KafkaTopicInitialConfiguration {

    private final KafkaConfig kafkaConfig;

    private final GenericApplicationContext context;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        KafkaAdmin kafkaAdmin = new KafkaAdmin(this.kafkaConfig.buildAdminProperties());
        kafkaAdmin.setFatalIfBrokerNotAvailable(true);
        return kafkaAdmin;
    }

    @PostConstruct
    public void init() {
        kafkaConfig.getDxTopics().forEach(dxTopic -> context.registerBean(dxTopic.getName(), NewTopic.class, dxTopic::toNewTopic));
    }

}
